package com.atguigu.flink.exec2;

import com.atguigu.flink.pojo.AdsClickLog;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * Created by Smexy on 2022/12/17
 *
 * 统计当天用户对广告的点击量，并统计拉黑的用户。
 *
 *    限制用户每一天对每个广告的点击次数为100次。
 *
 *      要求： 数据必须按照时间顺序有序。
 *
 *              用户1 在12-17 点击了 a广告  100
 *
 *              用户1 在12-18 点击了 a广告  1
 *
 *
 *
 *
 */
public class Demo4_AdsCLick
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

       // env.setParallelism(1);

        env
            .readTextFile("data/AdClickLog.csv").setParallelism(1)
            .map(new MapFunction<String, AdsClickLog>()
            {
                @Override
                public AdsClickLog map(String value) throws Exception {
                    String[] words = value.split(",");
                    return  new AdsClickLog(
                        Long.valueOf(words[0]),
                        Long.valueOf(words[1]),
                        words[2],
                        words[3],
                        Long.valueOf(words[4])
                    );

                }
            }).setParallelism(1)
            .keyBy(ads -> ads.getUserId() + "_" + ads.getAdsId())
            .process(new KeyedProcessFunction<String, AdsClickLog, String>()
            {
                //声明变量，当前用户对某个广告的点击次数
                //声明变量，代表是否已经拉黑，已经拉黑后续点击不统计
                //声明变量，用来标识日期的变化，如果日期变化就重置拉黑的状态及点击次数
                private ValueState<Integer> count;
                private ValueState<String> lastDate;
                private ValueState<Boolean> ifBlack;

                @Override
                public void open(Configuration parameters) throws Exception {
                    RuntimeContext context = getRuntimeContext();

                    ifBlack = context.getState(new ValueStateDescriptor<>("ifBlack", Boolean.class));
                    lastDate = context.getState(new ValueStateDescriptor<>("lastDate", String.class));
                    count = context.getState(new ValueStateDescriptor<>("count", Integer.class));
                }

                //状态： 算子中的非只读属性。 只读属性，无需声明为状态。
                private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");

                @Override
                public void processElement(AdsClickLog value, Context ctx, Collector<String> out) throws Exception {

                    String currentDate = this.format.format(new Date(value.getTimestamp() * 1000));
                    String lastDateStr = lastDate.value();
                    //判断当前是否已经跨天，如果跨天，一切统计要重置
                    if ( lastDateStr == null || !lastDateStr.equals(currentDate)){
                            ifBlack.update(false);
                            count.update(0);
                            lastDate.update(currentDate);
                    }

                    //累加
                    count.update(count.value() + 1);

                    Integer click = count.value();

                    //获取结果判断
                    if ( !ifBlack.value() && click >= 100 ){

                        //符合拉黑条件
                        out.collect(ctx.getCurrentKey() + ":"+ click + ",已拉黑");

                        ifBlack.update(true);

                    }else{

                        if (click < 100){
                            //正常条件
                            out.collect(ctx.getCurrentKey() + ":"+ click );
                        }

                    }

                }
            })
            .print().setParallelism(1);

        env.execute();

    }
}
